/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once
#include "common/config.h"
#include "defs.h"
#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "index/ix.h"
#include "index/ix_index_handle.h"
#include "record/rm_defs.h"
#include "system/sm.h"
#include "system/sm_meta.h"
#include <vector>


class SortExecutor : public AbstractExecutor {
   private:
    std::unique_ptr<AbstractExecutor> prev_;
    ColMeta cols_;                              // 框架中只支持一个键排序，需要自行修改数据结构支持多个键排序
    std::unique_ptr<RmRecord> current_tuple;

    [[maybe_unused]] bool is_end_ = false;

    std::vector<RmRecord> records_;
    int sum_records = 0;

    int idx = 0;
   public:
    // Constructor for SortExecutor class
    SortExecutor(std::unique_ptr<AbstractExecutor> prev, const TabCol &sel_cols, bool is_desc) :
        prev_(std::move(prev)) {
        cols_ = prev_->get_col_offset(sel_cols);    
        
        for(prev_->beginTuple(); !prev_->is_end(); prev_->nextTuple()) {
            records_.push_back(*prev_->Next());
            sum_records++;
        }  

        std::sort(records_.begin(), records_.end(), [is_desc, this](const RmRecord &a, const RmRecord &b) {
            std::vector<ColType> col_type = {cols_.type};
            std::vector<int> col_lens = {cols_.len};
            int res = ix_compare(a.data + cols_.offset, b.data + cols_.offset, col_type, col_lens);
            if(is_desc) res = -res;
            return res < 0;
        });
    }

    void beginTuple() override {
        idx = 0;
    }

    void nextTuple() override {
        idx ++;
    }

    std::unique_ptr<RmRecord> Next() override {
        return std::make_unique<RmRecord>(std::move(records_[idx]));
    }

    [[nodiscard]] bool is_end() const override {
        return idx == sum_records;
    }

    Rid &rid() override {
        return _abstract_rid;
    }

    [[nodiscard]] const std::vector<ColMeta> &cols() const override {
        return prev_->cols();
    };
};